package com.hbase.mr1;

import java.io.IOException;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/*
 * 1.MR读取hbase中t2表的部分数据，写入到t4表中
 *
 * 		Map -------- Reduce
 * 		读入输入目录中的数据（t2）-----map-----reduce-----写出到指定目录（t4）
 *
 * 2.目前读取的是hbase的数据，不是文本文件
 *   所以不能使用默认的TextFileInputFormat,自然RR的泛型无法确定！
 *
 *   需要先确定什么样的输入格式可以读取hbase中的表？
 *   	在hbase中，默认使用TableInputFormat输入格式
 *   		切片：一个region切一个
 *   		RR：RecordReader<ImmutableBytesWritable, Result>
 *   			ImmutableBytesWritable：rowkey
 *   			Result：一行数	据
 *
 * 3.输出的对象应该是一个Put对象！
 * 		在Mapper中，如果数据需要进行排序，必须作为key，如果不需要排序可以作为value
 *
 * 4.如果Mapper的数据类型是put类型，会自动设置Combiner！
 * 		无法避免设置Combiner，可以让Combiner的逻辑无法执行！
 * 		保证输出的key不能相等！
 *
 * 5.Hadoop MR处理的数据必须都是key-value
 * 		在Hadoop中，只要有Reducer，那么此时key-value必须实现序列化！
 * 		Writable：key实现WritableComparable
 * 				  value实现Writable
 *
 * 		为什么hbase中Put中没有实现Writable，程序依然没有报错呢？
 * 			因为：hbase自动提供了对Put类型的序列化器！
 *
 * 		之前强调key-value都必须实现Writable，是因为key-value实现了Writable
 * 		Hadoop为实现了Writable接口的类提供了序列化器！
 *
 * 		在Hadoop中，必须提供序列化器！至于类型是否实现序列化是无关紧要的！
 */
public class Example1Mapper extends TableMapper<Text, Put>{

	@Override
	protected void map(ImmutableBytesWritable key, Result value,
					   Mapper<ImmutableBytesWritable, Result, Text, Put>.Context context)
			throws IOException, InterruptedException {

		// 构建Put对象，封装rowkey
		Put put = new Put(key.copyBytes());

		Cell[] rawCells = value.rawCells();

		for (Cell cell : rawCells) {

			// 将读到的数据，进行过滤，只选择info:age=20的行的内容进行输出
			if(Bytes.toString(CellUtil.cloneFamily(cell)).equals("info")
					&& Bytes.toString(CellUtil.cloneQualifier(cell)).equals("age")
					&& Bytes.toString(CellUtil.cloneValue(cell)).equals("20")) {

				// 把符合要求的一行的所有单元格放入到put对象中
				for(Cell c : rawCells) {

					put.add(c);

				}

				// 把整个put写出
				context.write(new Text(Bytes.toString(key.copyBytes())), put);

			}

		}

	}

}















